# -*- coding: utf-8 -*-
"""
@Time: 2022/3/1 08:46
@Author: jins
@File: kafkaIndexFlink.py
@Introduction: 
"""
import json
from datetime import datetime, timedelta

from kafka import KafkaProducer

# producer = KafkaProducer(bootstrap_servers=['cdh01:9092', 'cdh02:9092', 'cdh03:9092'])
fmt_s = '%Y%m%d%H%M%S'
count = 1
for j in range(1, 11):
    device_id = str(10000000 + j)
    tm = datetime(2022, 10, 1, 0, 0)
    producer = KafkaProducer(bootstrap_servers=['cdh01:9092', 'cdh02:9092', 'cdh03:9092'])
    for i in range(2880):
        tm_res = str(tm.strftime(fmt_s))[::-1]
        row_key = tm_res + device_id + 'mi15'
        data = {"indexColumns": "device_id,data_time,frequency", "rowkey": row_key,
                "tableName": "ods_device_electric_ts"}
        send_message = json.dumps(data).encode()
        producer.send('kafka.index.flink', send_message)
        tm += timedelta(minutes=15)
        print(count)
        count += 1
